package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.storm.AbstractBolt;
import com.aotain.storm.TupleHelpers;

public class HurstBolt extends AbstractBolt{


	private static final long serialVersionUID = 1L;
	private OutputCollector collector = null;			
	//private HashMap<String,String> tripleAll = new HashMap<String,String>();
	private HashMap<String,Integer>pkgMap = new HashMap<String,Integer>();
	private HashMap<String,Integer> pvsumMap = new HashMap<String,Integer>();
	private HurstSliding cache;

	public HurstBolt(){		
		cache = new HurstSliding(60);
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("dip","hurst"));	
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60);
		return conf;
	}

	//计算需要的值
	private void countInLocal(Tuple tuple) {
		try
		{	
			Logger.getRootLogger().info("=====tuple====="+tuple);
			String dip = tuple.getStringByField("dip").trim();
			double uppkg = Double.parseDouble(tuple.getStringByField("uppacket").trim());
			double upoct = Double.parseDouble(tuple.getStringByField("upoct").trim());
			
			//包数和
			if(pkgMap.containsKey(dip)){
				int tmppv = pkgMap.get(dip) + (int)uppkg;
				pkgMap.put(dip, tmppv);
			}else{
				pkgMap.put(dip, (int)uppkg);
			}			
			//目的IP的连接数PV			
			int tmppv = 1;
			if(pvsumMap.get(dip) != null) {
				tmppv = pvsumMap.get(dip) + 1;
			}
			pvsumMap.put(dip, tmppv);
			
			//用于算法：上行流量除以上行包数
//			if(tripleAll.containsKey(dip)){				
//				String[] val = tripleAll.get(dip).split(",");	
//				double pkg = uppkg + Double.parseDouble(val[0]);
//				double oct = upoct + Double.parseDouble(val[1]);				
//				tripleAll.put(dip, pkg + "," + oct);
//			}else{
//				tripleAll.put(dip, uppkg + "," + upoct);
//			}
		}
		catch(Exception ex)
		{
			Logger.getRootLogger().error("=====countInLocal=========ERROR=================",ex);
		}
	}
	//以下是包数和除以连接次数
	private void emitCountingData(OutputCollector collector) {
		try{
			
		Logger.getRootLogger().info("======emitCountingData start========");
		
		if(pkgMap.size()==0 || pvsumMap.size()==0 ){
			Logger.getRootLogger().info(" >>>>>>>>>> pkgMap or pvsumMap is null");
			return;
		}
		
		HashMap<String,Double> infoCal = new HashMap<String,Double>();
		for (Entry<String, Integer> calEnt : pkgMap.entrySet()) {//192.168.1.1  10,200	
			String dip = calEnt.getKey();	
			if(!pvsumMap.containsKey(dip)) continue;
			double ret = calEnt.getValue()/pvsumMap.get(dip);
			Logger.getRootLogger().error("=====tripleAll=========loop====="+dip+">>>>"+calEnt.getValue()+">>>>"+ret);
			infoCal.put(dip, ret);			
		}
			
		cache.addPInfo(infoCal);
		Map<Integer,HashMap<String,Double>> infoMap = cache.getTuplesWindow();
		
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);		
		HashMap<String,String> dipHurst = new HashMap<String,String>();//192.168.1.1  1000|2000|3000
		
		for(HashMap<String,Double> singleMap: infoMap.values()) {
			if(singleMap == null) continue;
			for (Entry<String, Double> entry : singleMap.entrySet()) {
				if(dipHurst.containsKey(entry.getKey())){
					String tempcal = dipHurst.get(entry.getKey());					
					tempcal = tempcal + entry.getValue()+"|";					
					dipHurst.put(entry.getKey(), tempcal);
				}else{					
					if(!entry.getKey().isEmpty())
						dipHurst.put(entry.getKey(), entry.getValue()+"|");
					Logger.getRootLogger().info(">>>>>>>>>>dip:"+entry.getKey()+">>>callist:"+entry.getValue());	
				}
			}
		}
		
		for (Entry<String, String> hurstent : dipHurst.entrySet()) {			
			String[] arr = hurstent.getValue().split("\\|");				
			if(arr.length<5) continue;
			double hurstdata[] = new double[arr.length];
			for(int i=0;i<arr.length;i++){				
				if(arr[i].trim().isEmpty()) continue;					
				hurstdata[i] = Double.parseDouble(arr[i].trim());					
			}
			Logger.getRootLogger().info(">>>>>>>>>>dip-hurst:" + hurstent.getKey() + ">>>>>>>>>>" + hurstent.getValue());	
			com.aotain.common.Hurst H = new com.aotain.common.Hurst(hurstdata);
			double ret = H.Calculate();
			if(Double.isNaN(ret)) continue;
			
			Logger.getRootLogger().info(">>>>key" + hurstent.getKey() + ">>>>value>" +hurstent.getValue()+">>>hurst:"+ ret);
			String accesstime = new SimpleDateFormat("yyyyMMddHHmm").format(new Date(System.currentTimeMillis()+15000));
			String rowkey = String.format("%s_%s_pkgpv", hurstent.getKey(),accesstime);
			String tbName = "TEMP_DDOSHURST";			
	    	hbaseInstance.Add(tbName, rowkey, "cf", "ACCESSTIME", accesstime);
	    	hbaseInstance.Add(tbName, rowkey, "cf", "DIP", hurstent.getKey());
	    	hbaseInstance.Add(tbName, rowkey, "cf", "HURST",String.valueOf(ret));
			collector.emit(new Values(hurstent.getKey(),ret));
		
		}
		infoMap.clear();
		infoCal.clear();
		pvsumMap.clear();		
		pkgMap.clear();	
		cache.advanceHead();
		} catch (Exception e) {
			throw new FailedException("DosHurstBolt throws an exception!",e);
		}	
	}


	//以下是上行流量除以上行包书的算法
//	private void emitCountingData(OutputCollector collector) {
//		try{
//			
//		Logger.getRootLogger().info("======emitCountingData start========");
//		
//		if(tripleAll.size()==0){
//			Logger.getRootLogger().info(" >>>>>>>>>> tripleAll is null");
//			return;
//		}
//		Logger.getRootLogger().info(" >>>>>>>>>> tripleAll.size()" + tripleAll.size());
//		
//		HashMap<String,Double> infoCal = new HashMap<String,Double>();
//		for (Entry<String, String> calEnt : tripleAll.entrySet()) {//192.168.1.1  10,200	
//			String dip = calEnt.getKey();
//			Logger.getRootLogger().error("=====tripleAll=========loop====="+dip+">>>>"+calEnt.getValue());
//			String[] val = calEnt.getValue().split(",");
//			double ret = Double.parseDouble(val[1].trim())/Double.parseDouble(val[0].trim());
//			Logger.getRootLogger().error("=====tripleAll=========loop====="+dip+">>>>"+calEnt.getValue()+">>>>"+ret);
//			if(!infoCal.containsKey(dip)){
//				infoCal.put(dip, ret);
//			}
//		}
//			
//		cache.addPInfo(infoCal);
//		Map<Integer,HashMap<String,Double>> infoMap = cache.getTuplesWindow();
//		
//		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver,logserver);		
//		HashMap<String,String> dipHurst = new HashMap<String,String>();//192.168.1.1  1000|2000|3000
//		Logger.getRootLogger().info("loop start-----infoMap.size()--"+infoMap.size());	
//		
//		
//		for(HashMap<String,Double> singleMap: infoMap.values()) {
//			if(singleMap == null) continue;
//			for (Entry<String, Double> entry : singleMap.entrySet()) {
//				if(dipHurst.containsKey(entry.getKey())){
//					String tempcal = dipHurst.get(entry.getKey());					
//					tempcal = tempcal + entry.getValue()+"|";					
//					dipHurst.put(entry.getKey(), tempcal);
//				}else{					
//					if(!entry.getKey().isEmpty())
//						dipHurst.put(entry.getKey(), entry.getValue()+"|");
//					Logger.getRootLogger().info(">>>>>>>>>>dip:"+entry.getKey()+">>>callist:"+entry.getValue());	
//				}
//			}
//		}
//		
//		for (Entry<String, String> hurstent : dipHurst.entrySet()) {			
//			String[] arr = hurstent.getValue().split("\\|");				
//			if(arr.length<5) continue;
//			double hurstdata[] = new double[arr.length];
//			for(int i=0;i<arr.length;i++){				
//				if(arr[i].trim().isEmpty()) continue;					
//				hurstdata[i] = Double.parseDouble(arr[i].trim());					
//			}
//			Logger.getRootLogger().info(">>>>>>>>>>5555555:" + hurstent.getKey() + ">>>>>>>>>>" + hurstent.getValue());	
//			com.aotain.common.Hurst H = new com.aotain.common.Hurst(hurstdata);
//			double ret = H.Calculate();
//			if(Double.isNaN(ret)) continue;
//			
//			Logger.getRootLogger().info(">>>>key" + hurstent.getKey() + ">>>>value>" +hurstent.getValue()+">>>hurst:"+ ret);
//			String accesstime = new SimpleDateFormat("yyyyMMddHHmm").format(new Date(System.currentTimeMillis()+15000));
////			String rowkey = String.format("%s_%s_pkg", hurstent.getKey(),accesstime);
//			String rowkey = String.format("%s_%s_cal", hurstent.getKey(),accesstime);
//			String tbName = "TEMP_DDOSHURST";			
//	    	hbaseInstance.Add(tbName, rowkey, "cf", "ACCESSTIME", accesstime);
//	    	hbaseInstance.Add(tbName, rowkey, "cf", "DIP", hurstent.getKey());
//	    	hbaseInstance.Add(tbName, rowkey, "cf", "HURST",String.valueOf(ret));
//			collector.emit(new Values(hurstent.getKey(),ret));
//		
//		}
//		infoMap.clear();
//		infoCal.clear();
//		tripleAll.clear();		
//		cache.advanceHead();
//		} catch (Exception e) {
//			// TODO Auto-generated catch block
//			
//			throw new FailedException("DosHurstBolt throws an exception!",e);
//		}	
//	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}


	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			
			//if(isTickTuple(tuple)) {
			if(TupleHelpers.isTickTuple(tuple)) {
				emitCountingData(collector);  
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			collector.fail(tuple);
			throw new FailedException("DosHurstBolt throws an exception!");
		}	
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		this.collector = collector;
	}
	

}
